package com.zyh.kafka.day01;

import com.zyh.kafka.day02.MyPartitioner;
import com.zyh.kafka.day02.MyProducerInterceptor;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class KafkaProducerTest1 {
    public static void main(String[] args) {
        //1。提供生产者需要的配置参数
        Properties producerConfigs = new Properties();

        //a 链接的kafka集群的ip:端口
        producerConfigs.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"kafka24:9092");
        //b 设置消息-->byte[] 所需要的序列化工具 一个消息有key和value组成，key可以是不必须的
        producerConfigs.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        producerConfigs.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
        //c 设置自定义的分区类
        producerConfigs.setProperty(ProducerConfig.PARTITIONER_CLASS_CONFIG, MyPartitioner.class.getName());
        //d 设置自定义拦截器类
        producerConfigs.setProperty(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, MyProducerInterceptor.class.getName());

        //2。创建生产者
        KafkaProducer<String,String> kafkaProducer = new KafkaProducer<>(producerConfigs);

        //3。调用send方法发送消息
        for (int i = 0; i < 10; i++) {
            /*
                ProducerRecord(String topic,Integer partition,K key,V value)
                分区规则：
                    1，消息本身如果设置分区，则以设置的分区为主
                    2，如果没有分区，有key，则会以key的哈希值%分区数
                    3，没有分区，没有key，则会随机轮询
                        第一次发送消息，会产生一个随机值，随机值%可用分区数
                        从第二次开始，会在上一次key值的基础上+1，再%可用分区数
            */
            kafkaProducer.send(new ProducerRecord<>("topic-a", null, "hello", "hello-" + i));
        }

        //4。释放资源
        kafkaProducer.close();
    }
}
